package io.jnet.common.businessprocessor;

import java.util.Queue;
import java.util.concurrent.locks.LockSupport;
import com.jfireframework.baseutil.collection.buffer.ByteBuf;
import com.jfireframework.baseutil.concurrent.CpuCachePadingInt;
import com.jfireframework.baseutil.concurrent.MPSCQueue;
import io.jnet.common.api.AioListener;
import io.jnet.common.api.ChannelContext;
import io.jnet.common.streamprocessor.ProcesserUtil;
import io.jnet.common.streamprocessor.ProcessorTask;

public class ThreadAttachProcessor implements Runnable
{
    private final Queue<ProcessorTask> tasks          = new MPSCQueue<>();
    private static final int           IDLE           = 0;
    private static final int           WORK           = 1;
    private final CpuCachePadingInt    status         = new CpuCachePadingInt(WORK);
    private final AioListener          serverListener;
    private static final int           SPIN_THRESHOLD = 1 << 7;
    private int                        spin           = 0;
    private volatile Thread            owner;
    
    public ThreadAttachProcessor(AioListener serverListener)
    {
        this.serverListener = serverListener;
    }
    
    @Override
    public void run()
    {
        status.set(WORK);
        owner = Thread.currentThread();
        do
        {
            ProcessorTask task = tasks.poll();
            if (task == null)
            {
                spin = 0;
                for (;;)
                {
                    
                    if ((task = tasks.poll()) != null)
                    {
                        break;
                    }
                    else if ((spin += 1) < SPIN_THRESHOLD)
                    {
                        ;
                    }
                    else
                    {
                        spin = 0;
                        status.set(IDLE);
                        if ((task = tasks.poll()) != null)
                        {
                            status.set(WORK);
                            break;
                        }
                        else
                        {
                            LockSupport.park();
                            status.set(WORK);
                        }
                    }
                }
            }
            try
            {
                ChannelContext serverChannelContext = task.getChannelContext();
                if (serverChannelContext.isOpen())
                {
                    Object result = ProcesserUtil.process(serverChannelContext, serverChannelContext.inProcessors(), task.getData(), task.getInitIndex());
                    if (result instanceof ByteBuf<?>)
                    {
                        serverChannelContext.bufStorage().putBuf((ByteBuf<?>) result);
                        serverChannelContext.registerWrite();
                    }
                }
            }
            catch (Throwable e)
            {
                serverListener.catchException(e, task.getChannelContext());
            }
        } while (true);
    }
    
    public void commit(ProcessorTask task)
    {
        tasks.offer(task);
        tryExecute();
    }
    
    private void tryExecute()
    {
        int now = status.value();
        if (now == IDLE)
        {
            LockSupport.unpark(owner);
        }
    }
}
